package com.grj.payment.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.ExponentialBackOff;
import org.springframework.util.backoff.FixedBackOff;

import java.util.HashMap;
import java.util.Map;

/**
 * @author ganru
 * @date 2025/10/19 5:18
 */
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    // ConsumerFactory：可使用 JsonDeserializer for values if you use JSON
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "payment-service");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer");
        props.put("spring.json.trusted.packages", "*");
        return new DefaultKafkaConsumerFactory<>(props);
    }



    // DeadLetterPublishingRecoverer：失败后把消息发布到 DLQ topic
    @Bean
    public DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(KafkaTemplate<String, Object> template) {
        // 默认目标 topic: original-topic + ".DLT"
        return new DeadLetterPublishingRecoverer(template,
                (r, e) -> new org.apache.kafka.common.TopicPartition(r.topic() + ".DLT", r.partition()));
    }

    // ErrorHandler with backoff and retry attempts
    @Bean
    public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
        // ExponentialBackOff: initial interval 1s, multiplier 2.0, max interval 10s
        ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
        backOff.setMaxInterval(10000L);
//        FixedBackOff fixedBackOff = new FixedBackOff(2000L, 2L); // retry 2 times (3 total attempts)
//        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, fixedBackOff);

        // DefaultErrorHandler(deathLetterRecoverer, backOff)
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

        // Configure max attempts (how many polls will be retried before sending to DLQ)
        // Note: DefaultErrorHandler uses the BackOff to determine backoff between attempts.
        // To control attempts, you can set maxFailures if desired:
        // handler.setMaxFailures(3); // optional; DefaultErrorHandler handles attempts based on backOff and exception types

        // Optionally add classes to not retry (like IllegalArgumentException)
        // handler.addNotRetryableExceptions(IllegalArgumentException.class);

        return handler;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
            ConsumerFactory<String, Object> consumerFactory,
            DefaultErrorHandler errorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // 控制每次 poll 拉取记录的数目（可选）
        factory.setConcurrency(3); // 与 topic 分区数配合
        // 注入默认错误处理器
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}